﻿using Confluent.Kafka;
using KafKa.SchoolApi.Model;
using Microsoft.AspNetCore.Mvc;
using Newtonsoft.Json;
using RabbitMQ.Client;
using System.Text;

namespace KafKa.SchoolApi.Controllers
{
    /// <summary>
    /// 广播消息
    /// </summary>
    [ApiController]
    [Route("[controller]")]
    public class SchoolBroadcastController : ControllerBase
    {
        private readonly ILogger<SchoolBroadcastController> _logger;

        /// <summary>
        /// 
        /// </summary>
        public SchoolBroadcastController(ILogger<SchoolBroadcastController> logger)
        {
            _logger = logger;
        }

        /// <summary>
        /// 1、主题交换机(消息确认、消费一次)
        /// </summary>
        /// <param name="schoolNotices"></param>
        /// <returns></returns>
        [HttpPost("noticesCreate1")]
        public SchoolNotices CreateNotice1(SchoolNotices schoolNotices)
        {

            var factory = new ConnectionFactory
            {
                HostName = "localhost",
                Port = 5672,
                Password = "guest",
                UserName = "guest",
                VirtualHost = "/"
            };

            using (var connection = factory.CreateConnection())
            {
                var channel = connection.CreateModel();
                // 定义交换机
                channel.ExchangeDeclare(exchange: "teacher_topic", type: "topic");

                var productJson = JsonConvert.SerializeObject(schoolNotices);

                var message = Encoding.UTF8.GetBytes(productJson);
                //发送消息
                var properties = channel.CreateBasicProperties();
                //设置消息持久化
                properties.Persistent = true;
                channel.BasicPublish(exchange: "teacher_topic",
                                         routingKey: "sms.teacher_topic",
                                         basicProperties: properties,
                                         body: message);

            }
            _logger.LogInformation("发送教师广播...");
            return schoolNotices;
        }

        /// <summary>
        /// 2、生产者-固定分区发送
        /// </summary>
        /// <param name="schoolNotices"></param>
        /// <returns></returns>
        [HttpPost("noticesCreate2")]
        public SchoolNotices CreateNotice2(SchoolNotices schoolNotices)
        {

            var producerConfig = new ProducerConfig
            {
                BootstrapServers = "127.0.0.1:9092",
                MessageTimeoutMs = 50000
            };
            var builder = new ProducerBuilder<string, string>(producerConfig);
            using (var producer = builder.Build())
            {
                try
                {
                    var OrderJson = JsonConvert.SerializeObject(schoolNotices);
                    var dr = producer.ProduceAsync("teacher-notice2", new Message<string, string> { Key = "teacher", Value = OrderJson }).GetAwaiter().GetResult();
                    _logger.LogInformation("发送事件 {0} 到 {1} 成功", dr.Value, dr.TopicPartitionOffset);
                }
                catch (ProduceException<string, string> ex)
                {
                    _logger.LogError(ex, "发送事件到 {0} 失败，原因 {1} ", "teacher", ex.Error.Reason);
                }
            }
            return schoolNotices;

        }

        /// <summary>
        /// 2、生产者-固定分区发送
        /// </summary>
        /// <param name="schoolNotices"></param>
        /// <returns></returns>
        [HttpPost("noticesCreate3")]
        public SchoolNotices CreateNotice3(SchoolNotices schoolNotices)
        {
            //采用分区轮询算法，将消息分摊到队列中不同的分区
            #region 1、生产者 Producer
            {
                var producerConfig = new ProducerConfig
                {
                    BootstrapServers = "127.0.0.1:9092",
                    MessageTimeoutMs = 50000,
                    EnableIdempotence = true
                };

                var builder = new ProducerBuilder<string, string>(producerConfig);

                //消息生产者采用Hash轮询算法将消息分摊到不同的分区。
                builder.SetDefaultPartitioner(RoundRobinPartitioner);
                using (var producer = builder.Build())
                {
                    try
                    {
                        var OrderJson = JsonConvert.SerializeObject(schoolNotices);
                        TopicPartition topicPartition = new TopicPartition("create-notice-3", 2); // 指定分区发送消息
                        var dr = producer.ProduceAsync(topicPartition, new Message<string, string> { Key = "potion-1", Value = OrderJson }).GetAwaiter().GetResult();
                        //var dr = producer.ProduceAsync("order-create", new Message<string, string> { Key = "order-1", Value = OrderJson }).GetAwaiter().GetResult();
                        _logger.LogInformation("发送事件 {0} 到 {1} 成功", dr.Value, dr.TopicPartitionOffset);
                    }
                    catch (ProduceException<string, string> ex)
                    {
                        _logger.LogError(ex, "发送事件到 {0} 失败，原因 {1} ", "order", ex.Error.Reason);
                    }
                }
            }
            #endregion

            return schoolNotices;
        }

        /// <summary>
        /// 分区随机算法
        /// </summary>
        /// <param name="topic"></param>
        /// <param name="partitionCount"></param>
        /// <param name="keyData"></param>
        /// <param name="keyIsNull"></param>
        /// <returns></returns>
        private Partition RandomPartitioner(string topic, int partitionCount, ReadOnlySpan<byte> keyData, bool keyIsNull)
        {
            Random random = new Random();
            int partition = random.Next(partitionCount - 1);
            return new Partition(partition);
        }

        static int requestCount = 0;
        /// <summary>
        /// 分区轮询算法。
        /// 使用分区处理消息堆积
        /// </summary>
        /// <param name="topic"></param>
        /// <param name="partitionCount"></param>
        /// <param name="keyData"></param>
        /// <param name="keyIsNull"></param>
        /// <returns></returns>
        private Partition RoundRobinPartitioner(string topic, int partitionCount, ReadOnlySpan<byte> keyData, bool keyIsNull)
        {
            //使用静态变量累加，每次根据取模获取不同的分区
            int partition = requestCount % partitionCount;
            requestCount++;
            return new Partition(partition);
        }



    }
}
